package com.zengjianjun.canal.demo.annotation.processor;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.zengjianjun.canal.demo.annotation.CanalListener;
import com.zengjianjun.canal.demo.consumer.CanalConsumerInterface;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * canal自定义注解处理器
 *
 * @author zengjianjun
 */
@Slf4j
@Component
public class CanalListenerAnnotationProcessor {

    @Resource
    private ApplicationContext applicationContext;

    /**
     * 储存canal的连接信息
     * key：hostName+port+destination+subscribe
     * value：CanalConnector
     */
    private final static ConcurrentHashMap<String, CanalConnector> CANAL_CONNECTOR_MAP = new ConcurrentHashMap<>();

    /**
     * 创建一个 PriorityBlockingQueue 用于按照 batchId 排序批次数据
     */
    private static final PriorityBlockingQueue<AckBatchIdInfo> BATCH_ID_QUEUE = new PriorityBlockingQueue<>(100, Comparator.comparing(AckBatchIdInfo::getBatchId));

    public void processor() {
        List<Object> subclassList = this.findSubclasses(CanalConsumerInterface.class);
        if (CollectionUtils.isEmpty(subclassList)) {
            return;
        }
        subclassList.forEach(bean -> {
            for (Method method : bean.getClass().getDeclaredMethods()) {
                if (method.isAnnotationPresent(CanalListener.class)) {
                    CanalListener canalListener = method.getAnnotation(CanalListener.class);
                    String hostName = canalListener.hostName();
                    if (StringUtils.isEmpty(hostName)) {
                        hostName = CanalConsumerInterface.HOST_NAME;
                    }
                    int port = canalListener.port();
                    if (port == -1) {
                        port = CanalConsumerInterface.PORT;
                    }
                    String destination = canalListener.destination();
                    String subscribe = canalListener.subscribe();
                    int threadNum = canalListener.threadNum();
                    int batchSize = canalListener.batchSize();
                    int batchSizeTimeOut = canalListener.batchSizeTimeOut();
                    String mapKey = this.getCanalConnectorMapKey(hostName, port, destination, subscribe);
                    CanalConnector connector = CANAL_CONNECTOR_MAP.get(mapKey);
                    if (connector == null) {
                        connector = this.createCanalConnector(hostName, port, destination, subscribe);
                        CANAL_CONNECTOR_MAP.put(mapKey, connector);
                    }
                    CanalConnector finalConnector = connector;
                    // 消费消息
                    new Thread(() -> {
                        log.info("创建canal监听：[destination:{}], [subscribe:{}], [threadNum:{}], [batchSize:{}], [batchSizeTimeOut:{}]", destination, subscribe, threadNum, batchSize, batchSizeTimeOut);
                        this.canalMessageConsumer(bean, method, finalConnector, threadNum, batchSize, batchSizeTimeOut);
                    }).start();
                    // 消息确认
                    this.ackBatchId(threadNum, connector);
                }
            }
        });
    }

    private void canalMessageConsumer(Object bean, Method method, CanalConnector connector, int threadNum, int batchSize, int batchSizeTimeOut) {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        for (int i = 0; i < threadNum; i++) {
            executorService.execute(() -> {
                while (true) {
                    try {
                        Message message;
                        if (batchSizeTimeOut <= 0) {
                            message = connector.getWithoutAck(batchSize);
                        } else {
                            message = connector.getWithoutAck(batchSize, (long) batchSizeTimeOut, TimeUnit.SECONDS);
                        }
                        long batchId = message.getId();
                        if (batchId != -1 && !message.getEntries().isEmpty()) {
                            log.info("batchId = {}, size = {}", batchId, message.getEntries().size());
                            method.invoke(bean, message);
                            if (threadNum == 1) {
                                connector.ack(batchId);
                            } else {
                                // 多线程时需要按顺序ack确认提交batchId
                                BATCH_ID_QUEUE.add(AckBatchIdInfo.builder()
                                        .threadId(Thread.currentThread().getName())
                                        .batchId(batchId)
                                        .canalConnector(connector)
                                        .build());
                            }
                        }
                        Thread.sleep(50);
                    } catch (Exception e) {
                        log.error("canal get message error: ", e);
                    }
                }
            });
        }
    }

    /**
     * canal消息消费批次确认
     */
    public void ackBatchId(int threadNum, CanalConnector canalConnector) {
        // 当线程数为小于等于1时，在消费消息完成时，直接确认
        if (threadNum <= 1) {
            return;
        }
        new Thread(() -> {
            log.info("canal消息消费批次确认开始...");
            while (true) {
                try {
                    // 保证每个线程都有add数据进入BATCH_ID_QUEUE中，否则ack确认时可能顺序错乱，导致canal多线程消费异常
                    int threadGroupSize = Arrays.stream(BATCH_ID_QUEUE.toArray(new AckBatchIdInfo[0]))
                            .filter(data -> data.getCanalConnector().equals(canalConnector))
                            .collect(Collectors.groupingBy(AckBatchIdInfo::getThreadId)).size();
                    if (threadGroupSize < threadNum) {
                        Thread.sleep(300L);
                        continue;
                    }
                    // poll获取元素，take阻塞获取元素
                    Long batchId = BATCH_ID_QUEUE.take().getBatchId();
                    log.info("canal ack batchId = {}", batchId);
                    canalConnector.ack(batchId);
                } catch (Exception e) {
                    log.error("canal ack batchId error: ", e);
                }
            }
        }).start();
    }

    /**
     * 创建连接
     *
     * @param hostName    域名或ip
     * @param port        端口号
     * @param destination 实例
     * @param subscribe   订阅信息
     * @return 连接对象
     */
    private CanalConnector createCanalConnector(String hostName, int port, String destination, String subscribe) {
        log.info("创建canal连接: [hostName:{}], [port:{}], [destination:{}], [subscribe:{}]", hostName, port, destination, subscribe);
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(hostName, port),
                destination,
                "",
                ""
        );
        connector.connect();
        // 设置订阅规则
        connector.subscribe(subscribe);
        // 从头开始消费
        connector.rollback();
        return connector;
    }

    /**
     * 获取连接对象的map key
     *
     * @param hostName    域名或ip
     * @param port        端口号
     * @param destination 实例
     * @param subscribe   订阅信息
     * @return map key
     */
    private String getCanalConnectorMapKey(String hostName, int port, String destination, String subscribe) {
        StringBuilder sb = new StringBuilder();
        sb.append(hostName).append("-").append(port).append("-").append(destination);
        if (StringUtils.isNotEmpty(subscribe)) {
            sb.append("-").append(subscribe);
        }
        return sb.toString();
    }


    /**
     * 获取指定类的子类
     *
     * @param superClass 父类
     * @return 子类列表
     */
    public List<Object> findSubclasses(Class<?> superClass) {
        List<Object> subclasses = new ArrayList<>();
        Map<String, ?> beans = applicationContext.getBeansOfType(superClass);
        for (Object bean : beans.values()) {
            Class<?> beanClass = bean.getClass();
            if (superClass.isAssignableFrom(beanClass) && !superClass.equals(beanClass)) {
                subclasses.add(bean);
            }
        }
        return subclasses;
    }
}
